#include "AQbuffer.h"
#include "AQthreadpool.h"
#include "AQtimewheel.h"
#include <iostream>
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <fcntl.h>
#include <string>
#include <set>
#include <pthread.h>
#include <signal.h>
#include <stdlib.h>
#include <unordered_map>

using namespace AQBUFFER;
using namespace TIMEWHEEL;

#define EPOLLSIZE 50
#define IP "101.34.110.235"
#define PORT 9999
std::set<int> usrfd;
pthread_mutex_t set_mtx;
pthread_mutex_t tw_mtx;
pthread_mutex_t del;

const int beattime = 1;
timeWheel tw(60, beattime);
std::unordered_map<int, timer *> map;
const int timeslot = 100;//timeout

// set_none_block_mod
void setnbmod(int fd)
{
    int flag = fcntl(fd, F_GETFL, 0);
    fcntl(fd, F_SETFL, flag | O_NONBLOCK);
}

struct message
{
    const int servfd;
    int epfd;
};

// register new connection fd

void Send(int fd, BUFFER &buff)
{
    const void *p = buff.beginRead();
    // fixme
    send(fd, p, buff.readable(), 0);
}

void tw_update(timer *Timer)
{
    pthread_mutex_lock(&del);
    tw.delTimer(Timer->slc, Timer);
    timer *tmp = tw.addTimer(Timer->fd, timeslot, Timer->func);
    map[tmp->fd] = tmp;
    pthread_mutex_unlock(&del);
}

void close_conn(int clntfd)
{
    pthread_mutex_lock(&tw_mtx);
    map.erase(clntfd);
    pthread_mutex_unlock(&tw_mtx);
    pthread_mutex_lock(&set_mtx);
    usrfd.erase(clntfd);
    pthread_mutex_unlock(&set_mtx);
    close(clntfd);
}

void callBack(int fd)
{
    close_conn(fd);
}

void rgst(void *arg)
{
    BUFFER buff;
    int client_sock;
    sockaddr_in clnt_addr;
    socklen_t addr_size = sizeof(clnt_addr);
    message tmp = *(message *)arg;
    client_sock = accept(tmp.servfd, (struct sockaddr *)&clnt_addr, &addr_size);
    std::cout << "client_sock= " << client_sock << std::endl;
    struct epoll_event event;
    event.events = EPOLLIN | EPOLLET | EPOLLONESHOT;
    event.data.fd = client_sock;
    //线程安全？
    epoll_ctl(tmp.epfd, EPOLL_CTL_ADD, client_sock, &event);
    sigset_t mask;
    sigemptyset(&mask);
    sigaddset(&mask, SIGALRM);
    pthread_sigmask(SIG_SETMASK, &mask, NULL);
    pthread_mutex_lock(&tw_mtx);
    std::function<void(int)> fn = std::bind(callBack, client_sock);
    timer *tp = tw.addTimer(client_sock, timeslot, fn);
    map[client_sock] = tp;
    pthread_mutex_unlock(&tw_mtx);

    pthread_mutex_lock(&set_mtx);
    usrfd.insert(client_sock);
    pthread_mutex_unlock(&set_mtx);

    std::string greet = "Wellcome to AQ chatroom!";
    buff.writestring(greet);
    Send(client_sock, buff);
    buff.fix_readIndex(greet.size() + sizeof(int));
}

void loop(void *arg)
{
    BUFFER buff;
    int clntfd = *(int *)arg;
    while (1)
    {
        ssize_t n = buff.readFd(clntfd);
        if (buff.readable() > sizeof(int))
        {
            pthread_mutex_lock(&tw_mtx);
            if(!map[clntfd])//如果连接已经断掉了
            {
                pthread_mutex_unlock(&tw_mtx);
                break;
            }
            tw_update(map[clntfd]);
            pthread_mutex_unlock(&tw_mtx);

            void *p = buff.beginRead();
            int len = 0;
            memcpy(&len, p, sizeof(int));
            buff.nToh(&len, sizeof(int));

            pthread_mutex_lock(&set_mtx);
            int usernum = usrfd.size();
            std::set<int>::iterator it;
            for (it = usrfd.begin(); it != usrfd.end(); it++)
            {
                if (*it != clntfd)
                {
                    // fixme
                    Send(*it, buff);
                    std::cout << "** send to ** " << *it << std::endl;
                }
            }
            buff.fix_readIndex(len + sizeof(int));
            pthread_mutex_unlock(&set_mtx);
        }
        if (n == 0)
        {
            close_conn(clntfd);
            break;
        }
    }
}

void Beats(int sig)
{
    tw.beats();
    alarm(beattime);
}

void TimeWheelOpen(void *)
{
    signal(SIGALRM, Beats);
    alarm(beattime);
}

int main()
{

    int serv_sock;
    struct sockaddr_in serv_adr;
    Threadpool *pool = threadpoolcreate(4, 10, 1000);
    serv_sock = socket(PF_INET, SOCK_STREAM, 0);
    serv_adr.sin_family = AF_INET;
    serv_adr.sin_addr.s_addr = htonl(INADDR_ANY);
    serv_adr.sin_port = htons(PORT);

    //初始化set锁
    pthread_mutex_init(&set_mtx, NULL);
    pthread_mutex_init(&tw_mtx, NULL);
    pthread_mutex_init(&del, NULL);

    if (bind(serv_sock, (struct sockaddr *)&serv_adr, sizeof(serv_adr)) == -1)
    {
        printf("bind error! \n");
        return 0;
    }

    if (listen(serv_sock, 10) == -1)
    {
        printf("listen error! \n");
        return 0;
    }

    struct epoll_event event;
    struct epoll_event *ep_events;
    int epfd, event_num = 0;

    epfd = epoll_create(EPOLLSIZE);
    ep_events = (struct epoll_event *)malloc(EPOLLSIZE * sizeof(struct epoll_event));
    event.events = EPOLLIN | EPOLLET;
    event.data.fd = serv_sock;
    epoll_ctl(epfd, EPOLL_CTL_ADD, serv_sock, &event);
    threadpooladd(pool, TimeWheelOpen, NULL);

    sigset_t mask;
    sigemptyset(&mask);
    sigaddset(&mask, SIGALRM);
    pthread_sigmask(SIG_SETMASK, &mask, NULL);
    while (1)
    {
        event_num = epoll_wait(epfd, ep_events, EPOLLSIZE, -1);
        std::cout << "***** " << event_num << std::endl;
        if (event_num == -1)
        {
            printf("epoll_wait() error!\n");
            break;
        }
        for (int i = 0; i < event_num; i++)
        {
            if (ep_events[i].data.fd == serv_sock)
            {
                message tmp{serv_sock, epfd};
                // Fix me use threadpooladd
                rgst((void *)&tmp);
                // threadpooladd(pool,rgst,(void*)&tmp);
            }
            else
            {
                int clntfd = ep_events[i].data.fd;
                threadpooladd(pool, loop, (void *)&clntfd);
            }
        }
    }
    free(ep_events);
    threadpooldestory(pool);
    close(serv_sock);
    return 0;
}